package org.liang.code.rocketMQ;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class 事务发送消息 {

    public static void main(String[] args) throws MQClientException {

        // 事务消息一定要使用TransactionMQProducer事务生产者创建
        TransactionMQProducer producer = new TransactionMQProducer("ts1");

        // 设置NameServer
        producer.setNamesrvAddr("127.0.0.1:9876");

        // CachedThreadPool线程池用于回查本地事务状态
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool(r -> new Thread(r, "check-transaction-thread"));

        // 将生产者与线程池绑定
        producer.setExecutorService(cachedThreadPool);

        // 绑定事务监听器，用于执行代码
        TransactionListener transactionListener = new OrderTransactionListenerImpl();
        producer.setTransactionListener(transactionListener);

        //启动生产者
        producer.start();

        // 创建消息对象
        Message msg = new Message("TTmusic","" ,"1030", "1030订单与明细的完整JSON数据（略）".getBytes());

        // 一定要调用sendMessageInTransaction发送事务消息
        // 参数1：消息对象
        // 参数2：其他参数，目前用不到
        producer.sendMessageInTransaction(msg, null);

        producer.shutdown();

    }
}
